package com.zongtui.webcrawler.mq.rocketmq;

import com.alibaba.rocketmq.client.exception.MQBrokerException;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.common.message.Message;
import com.alibaba.rocketmq.remoting.exception.RemotingException;

import java.util.concurrent.TimeUnit;

/**
 * RocketMQ提供者.
 * author cnJason
 */
public class Producer {


    private static Producer producer;

    private DefaultMQProducer dp;

    public DefaultMQProducer getDp() {
        return dp;
    }

    public void setDp(DefaultMQProducer dp) {
        this.dp = dp;
    }

    private static Producer getInstance(String producerGroup, String nameSrvAddr, String instanceName) throws MQClientException,InterruptedException {
        if (producer == null) {

            producer = new Producer();
            DefaultMQProducer dp = new DefaultMQProducer(producerGroup);
            dp.setNamesrvAddr(nameSrvAddr);
            dp.setInstanceName(instanceName);
            producer.setDp(dp);
            producer.getDp().start();
        }
        return producer;
    }


    public SendResult sendMessage(String topic, String tag, String key, String body) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        Message message = new Message(topic, tag, key, body.getBytes());
        SendResult sendResult = producer.getDp().send(message);
        return sendResult;
    }

    public void closeProducer() {
        producer.getDp().shutdown();
    }


    public static void main(String[] args) throws MQClientException,
            InterruptedException {
        /**
         * 一个应用创建一个Producer，由应用来维护此对象，可以设置为全局对象或者单例<br>
         * 注意：ProducerGroupName需要由应用来保证唯一<br>
         * ProducerGroup这个概念发送普通的消息时，作用不大，但是发送分布式事务消息时，比较关键，
         * 因为服务器会回查这个Group下的任意一个Producer
         */
        Producer producer = Producer.getInstance("ProducerGroupName", "192.168.230.128:9876", "Producer");


        /**
         * 下面这段代码表明一个Producer对象可以发送多个topic，多个tag的消息。
         * 注意：send方法是同步调用，只要不抛异常就标识成功。但是发送成功也可会有多种状态，<br>
         * 例如消息写入Master成功，但是Slave不成功，这种情况消息属于成功，但是对于个别应用如果对消息可靠性要求极高，<br>
         * 需要对这种情况做处理。另外，消息可能会存在发送失败的情况，失败重试由应用来处理。
         */
        for (int i = 0; i < 100; i++) {
            try {

                SendResult sendResult = producer.sendMessage("TopicTest1", "TagA", "OrderId001", "hello metaQ");
                System.out.println(sendResult);
            } catch (Exception e) {
                e.printStackTrace();
            }
            TimeUnit.MILLISECONDS.sleep(1000);
        }

        /**
         * 应用退出时，要调用shutdown来清理资源，关闭网络连接，从MetaQ服务器上注销自己
         * 注意：我们建议应用在JBOSS、Tomcat等容器的退出钩子里调用shutdown方法
         */
        producer.closeProducer();
    }
}
